package com.miku.mq.consume;

import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.miku.common.ErrorCode;
import com.miku.constant.AiConstant;
import com.miku.exception.BusinessException;
import com.miku.exception.ThrowUtils;
import com.miku.model.entity.Chart;
import com.miku.model.entity.TextRecord;
import com.miku.model.entity.TextTask;
import com.miku.model.enums.ChartStatusEnum;
import com.miku.model.enums.TextStatusEnum;
import com.miku.service.ChartService;
import com.miku.service.CreditService;
import com.miku.service.TextRecordService;
import com.miku.service.TextTaskService;
import com.miku.utils.SparkUtils;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Map;

import static com.miku.constant.AiConstant.CHAHE_KEY;
import static com.miku.constant.MessageConstant.BI_QUEUE;
import static com.miku.constant.MessageConstant.BI_TO_MD_QUEUE;

@Component
@Slf4j
public class MyMessageConsumer {
    /**
     * 接收消息
     * @param message
     * @param channel
     * @param deliverTag
     */

    @Autowired
    private ChartService chartService;
    @Autowired
    private SparkUtils sparkUtils;
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private CreditService creditService;
    @Resource
    private TextRecordService textRecordService;
    @Resource
    private TextTaskService textTaskService;

    @SneakyThrows
    @RabbitListener(queues = BI_QUEUE)
    public void receiveMessage(String message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliverTag){
        log.info("接收消息是:{}",message);
        if (StringUtils.isNotBlank(message)){
            //如果失败,消息拒绝
            channel.basicNack(deliverTag,false,false);
            log.info("消息为空拒绝接收,此消息正在被转发到死信队列中");
        }
        Long chartId=Long.parseLong(message);
        //查询chart表
        Chart chart = chartService.getById(chartId);
        if (chart==null){
           channel.basicNack(deliverTag,false,false);
           log.info("chartId为:{}的chart不存在",chartId);
           throw  new BusinessException(ErrorCode.PARAMS_ERROR,"chartId为:"+chartId+"的chart不存在");
        }
        String goal = chart.getGoal();
        String chartType = chart.getChartType();
        String csvData = chart.getChartData();

        StringBuilder userInput = new StringBuilder();
        userInput.append("分析需求:").append("\n");
        //拼接分析目标
        String userGoal = goal;
        if (StringUtils.isNotBlank(chartType)) {
            userGoal += ",请使用:" + chartType;
        }
        userInput.append(userGoal).append("\n");
        userInput.append("原始数据:").append("\n");
        userInput.append(csvData).append("\n");
        //将Chart表的状态改为执行中
        boolean update = chartService.update(Wrappers.<Chart>lambdaUpdate()
                .eq(Chart::getId, chart.getId())
                .set(Chart::getStatus, ChartStatusEnum.RUNNING.getValue()));
        if (!update) {
            handleChartErrorByMq(chartId, "更新图表状态为运行中失败");
            return;
        }
        String result = sparkUtils.synchronousCalls(AiConstant.MATH_MAP, userInput.toString());
        //对返回结果做拆分,按照5个中括号进行拆分
        String[] split = result.split("【【【【【");
        //拆分完还需要校验
        if (split.length < 3) {
            handleChartErrorByMq(chartId,"ai生成数据格式不正确");
            return;
        }
        String genChart = split[1].trim();
        String genResult = split[2].trim();

        //TODO: 使用redis缓存图表数据,提高查询效率
        chart.setExecMessage("图表生成成功！！！");
        chart.setStatus(ChartStatusEnum.SUCCEEd.getValue());
        chart.setGenChart(genChart);
        chart.setGenResult(genResult);
        //执行成功
        //修改数据库状态为执行成功
        boolean updateById = chartService.updateById(chart);
        if (!updateById) {
            handleChartErrorByMq(chartId,"更新图表状态为成功中失败");
        }
        //消费积分-每次使用消费1积分
        Boolean credits = creditService.consumeCredits(chart.getUserId(), -1L);
        ThrowUtils.throwIf(!credits,ErrorCode.OPERATION_ERROR,"你的积分不足");
        //存入缓存
        redisTemplate.opsForValue().set(CHAHE_KEY+chartId,JSONUtil.toJsonStr(chart));
       //如果任务执行成功,手动执行ack
        channel.basicAck(deliverTag,false);
    }

    /**
     * 负责处理用ai服务将文本转为md格式
     * @param message
     * @param channel
     * @param deliverTag
     */
    @SneakyThrows
    @RabbitListener(queues=BI_TO_MD_QUEUE,ackMode = "MANUAL")
    public void recevieMessageBiToMd(String message, Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliverTag){
        log.info("接收消息是:{}",message);
        //如果消息为空
        if (StringUtils.isBlank(message)){
            log.error("消息为空");
            channel.basicNack(deliverTag,false,false);
            throw new BusinessException(ErrorCode.PARAMS_ERROR,"消息为空");
        }
        //将消息转为map
        Map<String,String> map = JSONUtil.toBean(message, Map.class);
        Long textTaskId=Long.valueOf(map.get("textTaskId"));
        //查询textRecord表
        TextRecord textRecord = textRecordService.getOne(Wrappers.<TextRecord>
                lambdaQuery().eq(TextRecord::getTextTaskId,textTaskId));
        //如果textRecord为空,将状态改为失败
        if (textRecord == null) {
            handleTextRecordError(textTaskId, "textRecord为空,修改失败");
        }
        //文章内容
        String textContent = textRecord.getTextContent();
        //构建向ai提问的内容
        StringBuilder userInput = new StringBuilder();
        userInput.append("文章内容:").append(textContent).append("\n");
        userInput.append("请帮我将刚刚的文章转为markdown格式,并返回markdown格式的文章,不要有任何多余的内容");
        //调用ai服务
        String result = sparkUtils.synchronousCalls(AiConstant.TEXT_MAP, userInput.toString());
        //将结果保存到textRecord表中
        boolean update = textRecordService.update(Wrappers.<TextRecord>lambdaUpdate()
                .eq(TextRecord::getTextTaskId, textTaskId)
                .set(TextRecord::getGenTextContent, result)
                .set(TextRecord::getStatus, TextStatusEnum.SUCCEEd.getValue())
                .set(TextRecord::getExecMessage, "成功"));
        if (!update){
            handleTextRecordError(textTaskId,"调用ai服务textRecord表更新失败");
        }
       //任务执行完成手动确认
             channel.basicAck(deliverTag,false);
    }

    /**
     * 修改textTask表的状态为失败
     * @param taskId
     * @param errorMessagr
     */
    private void handleTextTaskError(Long taskId, String errorMessagr) {
           log.error("textTask为空,taskId:{}",taskId);
        textTaskService.update(Wrappers.<TextTask>lambdaUpdate()
                .eq(TextTask::getId, taskId)
                .set(TextTask::getStatus, TextStatusEnum.FAILED.getValue())
                .set(TextTask::getExecMessage, errorMessagr));

    }

    /**
     * 修改textRecord表的状态为失败
     * @param textTaskId
     */
    private void handleTextRecordError(Long textTaskId,String errorMessage) {
        log.error("textRecord为空,textTaskId:{}",textTaskId);
            textRecordService.update(Wrappers.<TextRecord>lambdaUpdate()
                    .eq(TextRecord::getTextTaskId, textTaskId)
                            .set(TextRecord::getExecMessage,errorMessage)
                    .set(TextRecord::getStatus, TextStatusEnum.FAILED.getValue()));

    }


    /**
     * 修改失败,改数据库的状态为失败
     */

    private void handleChartErrorByMq(Long chartId,String execMessage) {
        //修改数据库状态失败
        boolean update = chartService.update(Wrappers.<Chart>lambdaUpdate()
                .eq(Chart::getId, chartId)
                .set(Chart::getStatus, ChartStatusEnum.FAILED.getValue())
                .set(Chart::getExecMessage, execMessage));
        if (!update){
            log.error("更新图表状态为失败,失败,chartId:{}",chartId);
        }

    }
}
